package io.kiki.sba.registry.server.clientInterface.providedata;

import com.google.common.collect.Sets;
import io.kiki.sba.registry.common.model.ClientToClientInterfaceServerNodeId;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.metaserver.ClientManagerAddress;
import io.kiki.sba.registry.common.model.metaserver.ClientManagerAddress.AddressVersion;
import io.kiki.sba.registry.net.NetUtil;
import io.kiki.sba.registry.remoting.netty.Channel;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.connections.ConnectionsService;
import io.kiki.sba.registry.server.clientInterface.providedata.FetchClientOffAddressService.ClientOffAddressResp;
import io.kiki.sba.registry.server.clientInterface.providedata.FetchClientOffAddressService.ClientOffAddressStorage;
import io.kiki.sba.registry.server.clientInterface.registry.ClientManagerMetric;
import io.kiki.sba.registry.server.clientInterface.registry.Registry;
import io.kiki.sba.registry.server.shared.providedata.AbstractFetchPersistenceSystemProperty;
import io.kiki.sba.registry.server.shared.providedata.SystemDataStorage;
import io.kiki.sba.registry.store.api.meta.ClientManagerAddressRepository;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.LoopExecuteTask;
import io.kiki.sba.registry.util.StringUtil;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import static io.kiki.sba.registry.common.model.constants.ValueConstants.CLIENT_OFF;
import static io.kiki.sba.registry.server.clientInterface.registry.ClientManagerMetric.ADDRESS_LOAD_DELAY_HISTOGRAM;
import static io.kiki.sba.registry.server.clientInterface.registry.ClientManagerMetric.CLIENT_OFF_GAUGE;


public class FetchClientOffAddressService extends AbstractFetchPersistenceSystemProperty<ClientOffAddressStorage, ClientOffAddressResp> {

    private static final Logger logger = LoggerFactory.getLogger(AbstractFetchPersistenceSystemProperty.class);

    protected final ClientManagerProcessor clientManagerProcessor = new ClientManagerProcessor();
    protected final ClientOpenFailWatchDog clientOpenFailWatchDog = new ClientOpenFailWatchDog();

    @Autowired
    private Config config;

    @Autowired
    private ConnectionsService connectionsService;

    @Autowired
    private Registry sessionRegistry;

    @Autowired
    private ClientManagerAddressRepository clientManagerAddressRepository;

    public FetchClientOffAddressService() {
        super(ValueConstants.CLIENT_OFF_ADDRESS_DATA_ID, new ClientOffAddressStorage(INIT_VERSION, Collections.emptyMap(), Collections.emptySet(), null));
    }

    @PostConstruct
    public void postConstruct() {
        CLIENT_OFF_GAUGE.func(() -> storage.get().clientOffAddress.size());
        ConcurrentUtils.createDaemonThread("ClientManagerProcessor", clientManagerProcessor).start();
        ConcurrentUtils.createDaemonThread("clientOpenFailWatchDog", clientOpenFailWatchDog).start();
    }

    @Override
    protected int getSystemPropertyIntervalMillis() {
        return config.getClientManagerAddressIntervalMillis();
    }

    @Override
    protected ClientOffAddressResp fetchFromPersistence() {
        ClientManagerAddress clientManagerAddress = clientManagerAddressRepository.queryClientOffData();
        return new ClientOffAddressResp(clientManagerAddress.getVersion(), clientManagerAddress.getClientOffAddress(), clientManagerAddress.getReduces());
    }

    @Override
    protected boolean doProcess(ClientOffAddressStorage expect, ClientOffAddressResp data) {
        if (expect.updating.get() != null) {
            return false;
        }
        Set<String> news = data.clientOffAddress.keySet();
        Set<String> olds = expect.clientOffAddress.keySet();
        long oldVersion = expect.getVersion();

        Set<String> toBeRemove = Sets.difference(olds, news);
        Set<String> toBeClientOpen = Sets.difference(toBeRemove, data.reduces);
        Set<String> toBeAdd = Sets.difference(news, olds);

        Map<String, AddressVersion> adds = new HashMap<>(toBeAdd.size());
        for (String addAddress : toBeAdd) {
            AddressVersion add = data.clientOffAddress.get(addAddress);
            adds.put(addAddress, add);
        }
        StringUtil.checkEquals(toBeAdd.size(), adds.size(), "toBeAdd");
        try {

            ClientOffAddressStorage update = new ClientOffAddressStorage(data.getVersion(), data.clientOffAddress, data.reduces, new ClientOffTable(toBeAdd, toBeRemove, toBeClientOpen));
            if (!compareAndSet(expect, update)) {
                logger.error("update clientOffAddress:{} fail.", update.getVersion());
                return false;
            }
        } catch (Throwable t) {
            logger.error("update clientOffAddress:{} error.", data, t);
        }

        afterPropertySet(oldVersion, adds);
        logger.info("olds clientOffAddress:{}, oldVersion:{}, toBeAdd.size:{}, toBeAdd:{}, toBeClientOpen.size:{}, toBeClientOpen:{}, current clientOffAddress:{}, newVersion:{}", olds, oldVersion, toBeAdd.size(), toBeAdd, toBeClientOpen.size(), toBeClientOpen, storage.get().clientOffAddress.keySet(), storage.get().getVersion());
        return true;
    }

    private void afterPropertySet(long oldVersion, Map<String, AddressVersion> adds) {
        if (oldVersion == INIT_VERSION) {
            return;
        }
        long load = System.currentTimeMillis();
        for (AddressVersion value : adds.values()) {
            long cost = load - value.getVersion();
            ADDRESS_LOAD_DELAY_HISTOGRAM.observe(cost);
            logger.info("[LoadDelay]address:{}, version:{}, loaded:{}, cost:{}", value.getAddress(), value.getVersion(), load, cost);
        }
    }

    boolean processUpdating() {
        final ClientOffTable table = storage.get().updating.getAndSet(null);
        if (table == null) {
            return true;
        }

        Set<String> addAddress = table.addAddress;
        Set<String> removes = table.removes;
        Set<String> clientOpens = table.clientOpens;

        if (CollectionUtils.isEmpty(addAddress) && CollectionUtils.isEmpty(removes) && CollectionUtils.isEmpty(clientOpens)) {
            return true;
        }

        if (CollectionUtils.isNotEmpty(addAddress)) {
            doTrafficOff(addAddress);
        }

        if (CollectionUtils.isNotEmpty(removes)) {
            unMarkChannel(removes);
        }

        if (CollectionUtils.isNotEmpty(clientOpens)) {
            doTrafficOn(clientOpens);
        }
        return true;
    }

    private void unMarkChannel(Set<String> ipSet) {
        connectionsService.markChannelAndGetIpConnects(ipSet, CLIENT_OFF, null);
    }

    private void doTrafficOff(Set<String> ipSet) {
        List<ClientToClientInterfaceServerNodeId> conIds = connectionsService.markChannelAndGetIpConnects(ipSet, CLIENT_OFF, Boolean.TRUE);

        if (CollectionUtils.isEmpty(conIds)) {
            return;
        }
        logger.info("clientOff conIds: {}", conIds.toString());
        sessionRegistry.clientOff(conIds);
    }

    private void doTrafficOn(Set<String> ipSet) {
        List<String> connections = connectionsService.closeIpConnects(new ArrayList<>(ipSet));
        if (CollectionUtils.isNotEmpty(connections)) {
            ClientManagerMetric.CLIENT_OPEN_COUNTER.inc(connections.size());
            logger.info("clientOpen conIds: {}", connections);
        }
    }

    protected void processClientOpen() {
        List<io.kiki.sba.registry.api.Channel> channels = connectionsService.getAllChannel();
        ClientOffAddressStorage storage = this.storage.get();
        Map<String, AddressVersion> clientOffAddress = storage.clientOffAddress;
        Set<String> reduces = storage.reduces;

        Set<String> retryClientOpen = new HashSet<>(8);
        for (io.kiki.sba.registry.api.Channel channel : channels) {
            String key = NetUtil.toAddressString(channel.getRemoteAddress());
            String ip = connectionsService.getIpFromConnectId(key);

            Channel boltChannel = (Channel) channel;
            Object value = boltChannel.getConnAttribute(CLIENT_OFF);
            if (Boolean.TRUE.equals(value) && !clientOffAddress.containsKey(ip)) {

                if (reduces.contains(ip)) {
                    unMarkChannel(Collections.singleton(ip));
                } else {
                    logger.warn("[ClientOpenFail] ip:{} client open fail.", ip);
                    retryClientOpen.add(ip);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(retryClientOpen)) {
            logger.info("[ClientOpenRetry] ips:{} retry client open.", retryClientOpen);
            doTrafficOn(retryClientOpen);
        }
    }

    public Set<String> getClientOffAddress() {
        return storage.get().clientOffAddress.keySet();
    }

    public AddressVersion getAddress(String address) {
        return storage.get().clientOffAddress.get(address);
    }

    public long lastLoadVersion() {
        return storage.get().getVersion();
    }

    public void wakeup() {
        watchDog.wakeup();
    }


    public FetchClientOffAddressService setSessionServerConfig(Config config) {
        this.config = config;
        return this;
    }


    public FetchClientOffAddressService setConnectionsService(ConnectionsService connectionsService) {
        this.connectionsService = connectionsService;
        return this;
    }


    public FetchClientOffAddressService setSessionRegistry(Registry sessionRegistry) {
        this.sessionRegistry = sessionRegistry;
        return this;
    }


    public ClientOffAddressStorage getStorage() {
        return storage.get();
    }

    protected static class ClientOffAddressStorage extends SystemDataStorage {
        final Map<String, AddressVersion> clientOffAddress;

        final Set<String> reduces;

        final AtomicReference<ClientOffTable> updating;

        public ClientOffAddressStorage(long version, Map<String, AddressVersion> clientOffAddress, Set<String> reduces, ClientOffTable updating) {
            super(version);
            this.clientOffAddress = clientOffAddress;
            this.reduces = reduces;
            this.updating = new AtomicReference<>(updating);
        }
    }

    protected static class ClientOffAddressResp extends SystemDataStorage {
        final Map<String, AddressVersion> clientOffAddress;

        final Set<String> reduces;

        public ClientOffAddressResp(long version, Map<String, AddressVersion> clientOffAddress, Set<String> reduces) {
            super(version);
            this.clientOffAddress = clientOffAddress;
            this.reduces = reduces;
        }
    }

    static final class ClientOffTable {
        final Set<String> addAddress;

        final Set<String> removes;

        final Set<String> clientOpens;

        public ClientOffTable(Set<String> addAddress, Set<String> removes, Set<String> clientOpens) {
            this.addAddress = addAddress;
            this.removes = removes;
            this.clientOpens = clientOpens;
        }
    }

    protected final class ClientManagerProcessor extends LoopExecuteTask {

        @Override
        public void _execute_() {
            processUpdating();
        }

        @Override
        public void _wait_() {
            int clientManagerIntervalMillis = config.getClientManagerIntervalMillis();
            ConcurrentUtils.sleepUninterruptibly(clientManagerIntervalMillis, TimeUnit.MILLISECONDS);
        }
    }

    protected final class ClientOpenFailWatchDog extends LoopExecuteTask {

        @Override
        public void _execute_() {
            processClientOpen();
        }

        @Override
        public void _wait_() {
            int clientOpenIntervalSecs = config.getClientOpenIntervalSecs();
            ConcurrentUtils.sleepUninterruptibly(clientOpenIntervalSecs, TimeUnit.SECONDS);
        }
    }
}
